package com.star.sync.elasticsearch.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.github.pagehelper.PageHelper;
import com.star.sync.elasticsearch.dao.UserDao;
import com.star.sync.elasticsearch.model.User;
import com.star.sync.elasticsearch.model.UserVo;
import com.star.sync.elasticsearch.service.UserService;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

/**
 * @author: zhoujunming
 * @Description: TODO
 * @date: 2023/1/10 16:54
 * @email: zhoujunming@bluemoon.com.cn
 */
@Service
public class UserServiceImpl implements UserService {
    @Autowired
    @Qualifier("client")
    private RestHighLevelClient restHighLevelClient;

    private static final String USER_INDEX = "user";

    private static final int START_OFFSET = 0;

    private static final  int MAX_COUNT = 1000;

    @Autowired
    private UserDao userDao;

    @Override
    public void insert(User user) {
        userDao.insert(user);
    }

    @Override
    public List<User> searchTerm(String key, String value) throws IOException {
        SearchRequest searchRequest = new SearchRequest(USER_INDEX);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.termQuery(key,value));
        searchSourceBuilder.from(START_OFFSET);
        searchSourceBuilder.size(MAX_COUNT);
        searchRequest.source(searchSourceBuilder);
        searchSourceBuilder.trackTotalHits(true);
        SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        System.out.println(JSONObject.toJSON(response));

        SearchHit[] hits = response.getHits().getHits();
        List<User> userList = new LinkedList<>();
        for(SearchHit hit: hits){
            User user = JSONObject.parseObject(hit.getSourceAsString(),User.class);
            userList.add(user);
        }
        return userList;
    }

    @Override
    public List<User> searchName(String name) {
        return userDao.searchName(name);
    }

    @Override
    public List<User> searchDTerm(String key,String value) throws IOException{
        SearchRequest searchRequest = new SearchRequest(USER_INDEX);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //Range 查找
        RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(key);
        rangeQueryBuilder.gte(value);
        searchSourceBuilder.query(rangeQueryBuilder);
        searchSourceBuilder.from(START_OFFSET);
        searchSourceBuilder.size(MAX_COUNT);
        searchRequest.source(searchSourceBuilder);
        searchSourceBuilder.trackTotalHits(true);
        SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        System.out.println(JSONObject.toJSON(response));

        SearchHit[] hits = response.getHits().getHits();
        List<User> userList = new LinkedList<>();
        for(SearchHit hit: hits){
            User user = JSONObject.parseObject(hit.getSourceAsString(),User.class);
            userList.add(user);
        }
        return userList;
    }

    @Override
    public List<User> searchPageDB(Integer pageSize, Integer pageNo) {
        PageHelper.startPage(pageNo, pageSize);
        // 不带分页的查询
        List<User> list = userDao.searchPageDB();
        // 可以将结果转换为 Page , 然后获取 count 和其他结果值
        com.github.pagehelper.Page listWithPage = (com.github.pagehelper.Page) list;
        return listWithPage.getResult();
    }

    @Override
    public List<User> searchPageES(Integer pageSize, Integer pageNo) throws IOException {
        SearchRequest searchRequest = new SearchRequest(USER_INDEX);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //分页查询
        int pageCount = pageNo==1?pageNo:pageNo*pageSize;
        searchSourceBuilder.from(pageCount);
        searchSourceBuilder.size(pageSize);
        //排序条件
        searchSourceBuilder.sort("id", SortOrder.ASC);
        searchRequest.source(searchSourceBuilder);
        SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        SearchHit[] hits = response.getHits().getHits();
        List<User> userList = new LinkedList<>();
        for(SearchHit hit: hits){
            User user = JSONObject.parseObject(hit.getSourceAsString(),User.class);
            userList.add(user);
        }
        return userList;
    }

    @Override
    public User searchDetailDB(Integer id) {
        return userDao.searchDetailDB(id);
    }

    @Override
    public int searchCountDB(){
        return userDao.searchCountDB();
    }

    @Override
    public Long searchCountES() throws IOException {
        CountRequest countRequest = new CountRequest();
        // 绑定索引名
        countRequest.indices(USER_INDEX);
        CountResponse response = restHighLevelClient.count(countRequest, RequestOptions.DEFAULT);
        return response .getCount();
    }

    @Override
    public List<User> searchGroupDB() {
        return userDao.searchGroupDB();
    }

    @Override
    public List<UserVo> searchGroupES() throws IOException {
       /* SearchRequest searchRequest = new SearchRequest(USER_INDEX);
        SearchSourceBuilder builder = new SearchSourceBuilder();
        builder.size(0);
        AggregationBuilder aggregationBuilder = AggregationBuilders.terms("ageGroup").field("age").order(BucketOrder.count(false));
        builder.aggregation(aggregationBuilder);
        searchRequest.source(builder);
        SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        // 获取数据
        Aggregations aggregations = response.getAggregations();
        ParsedStringTerms ageGroup = aggregations.get("ageGroup");
        List<? extends Terms.Bucket> buckets = ageGroup.getBuckets();
        List<UserVo> userList = new ArrayList<>();
        for (Terms.Bucket bucket : buckets) {
            UserVo userVo = new UserVo();
            userVo.setAge(Integer.parseInt(bucket.getKey().toString()));
            userVo.setCount(bucket.getDocCount());
            userList.add(userVo);
        }*/
        List<UserVo> userList = new ArrayList<>();
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.size(0);
        AggregationBuilder aggregationBuilder = AggregationBuilders.terms("age");
        searchSourceBuilder.aggregation(aggregationBuilder);
        SearchRequest searchRequest = new SearchRequest(USER_INDEX);
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        Terms terms = searchResponse.getAggregations().get("agg");
        for (Terms.Bucket entry : terms.getBuckets()) {
            UserVo userVo = new UserVo();
            userVo.setAge(Integer.parseInt(entry.getKey().toString()));
            userVo.setCount(entry.getDocCount());
            userList.add(userVo);
        }
        return userList;
    }

    @Override
    public List<User> searchPageScrollES(Integer pageSize, Integer pageNo) throws IOException {
        SearchRequest searchRequest = new SearchRequest(USER_INDEX);

        //失效时间为1min
        Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1));
        //封存快照
        searchRequest.scroll(scroll);
        /**
         * 查询条件
         */
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
//        TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery("tag", "疫情");
//        searchSourceBuilder.query(termQueryBuilder);
        /**
         * 分页参数
         */
        searchSourceBuilder.size(pageSize);

        //放入文档中
        searchRequest.source(searchSourceBuilder);

        //远程查询
        SearchResponse searchResponse =restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        //元素数量
        Iterator<SearchHit> it1 = searchResponse.getHits().iterator();
        while (it1.hasNext()) {
            SearchHit next = it1.next();
        }
        //计算总数量
        long totalCount = searchResponse.getHits().getTotalHits().value;
        //得到总页数
        int page = (int) Math.ceil((float) totalCount / 2);
        //多次遍历分页，获取结果
        String scrollId = searchResponse.getScrollId();
        List<User> userList = new ArrayList<>();
        for (int i = 1; i <= page; i++) {
            //获取到该id
            SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
            searchScrollRequest.scroll(scroll);
            SearchResponse response = restHighLevelClient.scroll(searchScrollRequest, RequestOptions.DEFAULT);
            //打印数据
            SearchHits hits = response.getHits();
            scrollId = response.getScrollId();
            Iterator<SearchHit> iterator = hits.iterator();
            while (iterator.hasNext()) {
                User user = JSONObject.parseObject(iterator.next().getSourceAsString(),User.class);
                userList.add(user);
            }
        }
        return userList;
    }
}
